
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;


import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

public class AvgSource {
    public static class AvgSourceMap extends Mapper<LongWritable, Text, Text, FlowBan> {

         Text text = new Text();
        protected void map(LongWritable key, Text value, Context context) throws java.io.IOException, InterruptedException {
            String line = value.toString();
            String[] split = line.split("   ");
            String phone = split [1];
            int upFlow = Integer.parseInt(split[split.length-2]);
            int dFlow = Integer.parseInt(split[split.length-1]);
            text.set(phone);
            context.write(text,new FlowBan(phone,upFlow,dFlow));
        }

        public static class AvgSourceReduce extends Reducer<Text, FlowBan, Text, FlowBan> {

            protected void reduce(Text key, Iterable<FlowBan> value, Context context) throws java.io.IOException, InterruptedException {
                int count = 0;
                int index = 0;
                for (FlowBan flowBan : value) {
                    count = count + flowBan.getdFlow();
                    index = index + flowBan.getUpFlow();
                }
                context.write(key, new FlowBan(key.toString(),count,index));

            }
        }

        public static void main(String[] args) {
            Configuration conf = new Configuration();

            try {
                //新建一个Job工作
                Job job = new Job(conf);
                //设置运行类
                job.setJarByClass(AvgSource.class);
                //设置要执行的mapper类（自己书写的）
                job.setMapperClass(AvgSourceMap.class);
                //设置要执行的reduce类（自己书写的）
                job.setReducerClass(AvgSourceReduce.class);
                //设置输出key的类型
                job.setMapOutputKeyClass(Text.class);
                //设置输出value的类型
                job.setMapOutputValueClass( FlowBan.class);

                File file = new File("d:/file/output");
                if (file.exists()){
                    FileUtil.fullyDelete(file);
                }



                //mapreduce 输入数据的文件/目录
                FileInputFormat.addInputPath(job, new Path("d:/file/input"));
                //mapreduce 执行后输出的数据目录
                FileOutputFormat.setOutputPath(job, new Path("d:/file/output"));
                //设置ruduce任务的个数，默认个数为一个（一般reduce的个数越多效率越高）
                job.setNumReduceTasks(1);
                //执行完毕退出
                System.exit(job.waitForCompletion(true) ? 0 : 1);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }


    }

}
